package org.wsocket.socketset;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Timer;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;
import org.wsocket.WSocket;
import org.wsocket.Exception.WSocketException;
import org.wsocket.connector.Connector;
import org.wsocket.message.Message;
import org.wsocket.thread.ReceiveMessageTask;
import org.wsocket.thread.SendMessageTask;
import org.wsocket.thread.ThreadPool;
import org.wsocket.timer.TimeOutTimerTask;


public class SocketSet extends Thread {
	
	Logger log=Logger.getLogger(this.getClass());
	private ConcurrentHashMap<String, Connector> connectors = new ConcurrentHashMap<String, Connector>();
	private ConcurrentHashMap<String, String> c_connectors = new ConcurrentHashMap<String, String>();
	private ConcurrentHashMap<String, SelectionKey> c_sk = new ConcurrentHashMap<String, SelectionKey>();
	
	private LinkedList<Connector> pending = new LinkedList<Connector>();

	private Selector selector = null;
	Timer timeOutTimer = new Timer();
	
	private String key = null;

	private ArrayList<Callable<Boolean>> tasks = new ArrayList<Callable<Boolean>>();

	public String getKey() {
		return key;
	}

	public SocketSet(String key) {
		try {
			this.selector = Selector.open();
			this.key = key;
			timeOutTimer.schedule(
					new TimeOutTimerTask(this.connectors,this.c_connectors, this.c_sk), 
					WSocket.getDefaultTimeOut(), 
					WSocket.getDefaultTimeOut());
		} catch (IOException e) {
			log.error(e);
		}
	}

	public void addConnector(Connector connector) throws IOException {
		synchronized (pending) {
			pending.add(connector);
			c_connectors.put(connector.getKey(), "");
		}
		selector.wakeup();
	}

	public Connector getConector(String key) {
		return connectors.get(key);
	}

	public void broadCast(String message, String fromKey)
			throws WSocketException {
		for (String key : connectors.keySet()) {
			if (fromKey != null && key.equals(fromKey)) {
				continue;
			}
			Connector target = connectors.get(key);
			SendMessageTask task = new SendMessageTask(new Message(null,
					target, message));
			ThreadPool.assginTask(task);
		}
	}

	public void run() {
		if (log.isDebugEnabled()) {
			log.debug("new selector run.....");
		}
		while (true) {
			try {
				if (selector.select() > 0) {
					processSelectedKeys();
				}
				processPendingTargets();
			} catch (IOException x) {
				log.error(x);
			} catch (WSocketException e) {
				log.error(e);
			}
		}
	}

	// Process any targets in the pending list
	private void processPendingTargets() throws WSocketException {
		synchronized (pending) {
			while (pending.size() > 0) {
				Connector t = (Connector) pending.removeFirst();
				try {
					SelectionKey sk = t.getSocketChannel().register(selector,
							SelectionKey.OP_READ, t);
					c_sk.put(t.getKey(), sk);
					connectors.put(t.getKey(), t);
					if (log.isDebugEnabled()) {
						log.debug("selector inner size--"
								+ selector.keys().size());
					}
					WSocket.getListener().onConnect(t);
				} catch (IOException x) {
					log.error(x);
					throw new WSocketException(x);
				}
			}
		}
	}

	// Process keys that have become selected
	private void processSelectedKeys() throws IOException, WSocketException {
		tasks.clear();
		Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
		while (keyIter.hasNext()) {
			SelectionKey sk = keyIter.next();
			keyIter.remove();
			if (sk.isValid() && sk.isReadable() && sk.channel().isOpen()) {
				c_connectors.put(((Connector) sk.attachment()).getKey(), "");
				ReceiveMessageTask task = new ReceiveMessageTask(sk, connectors);
				tasks.add(task);
			}
		}
		if(tasks.size()>0){
			ThreadPool.assginTasks(tasks);
		}
	}

	public void finalize() {
		try {
			this.selector.close();
		} catch (IOException ex) {
			log.error(ex);
		}
	}

}
